#include "RecvMessThread.h"
#include "Log.h"
#include "common.h"
#include <sys/time.h>
#include "RingBuf.h"

int g_sleep = 10 ;//(1000.00 / double(config.bandWidth / config.maxDataSize))+0.5;
extern int g_nCounter;
int g_nMaxCounter  = 1000;//g_nMaxCounter = config.maxCounter;
///////////////////////////////////////////////////////////////
CRecvMessThread::CRecvMessThread() 
{
	g_nCounter = 0;
	bzero(&ev , sizeof(ev));
	epfd = epoll_create(5000);
	m_data = NULL;
	m_data = new CData();
}

CRecvMessThread::~CRecvMessThread()
{
	close(epfd);
	if(m_data)
	{
		delete m_data;
		m_data = NULL;
	}
}

unsigned int CRecvMessThread::getCounter()
{
	return g_nCounter;
}

bool CRecvMessThread::setNonBlock(int sockfd)
{
	int opts = fcntl(sockfd , F_GETFL);
	if(-1 == opts)
	{
		debug_output("%s\n" , "fcntl F_GETFL is faild");
		return false;
	}

	opts = opts | O_NONBLOCK;
	if(fcntl(sockfd , F_SETFL , opts) < 0)
	{
		debug_output("%s\n" , "fcntl F_SETFL is faild");
		return false;
	}
	return true;
}

void CRecvMessThread::addSocket(int nSocket)
{
	//struct epoll_event ev;

	setNonBlock(nSocket);
	ev.data.fd = nSocket;
	ev.events = EPOLLIN|EPOLLET|EPOLLERR|EPOLLHUP;
	epoll_ctl(epfd , EPOLL_CTL_ADD , nSocket , &ev);
	
//	debug_output("addEpoll ............socket = %d, m_ncounter = %d\n", nSocket, g_nCounter);
}
void CRecvMessThread::modEpoll(int nSocket, unsigned int event)
{
//	struct epoll_event ev;
//	bzero(&ev, sizeof(ev));
	ev.data.fd = nSocket;
	ev.events = event| EPOLLET;
	epoll_ctl(epfd, EPOLL_CTL_MOD, nSocket, &ev);
}
void CRecvMessThread::delEpoll(int nSocket)
{
	//struct epoll_event ev;
//	bzero(&ev, sizeof(ev));
	ev.data.fd = nSocket;
	epoll_ctl(epfd, EPOLL_CTL_DEL, nSocket, &ev);
	close(nSocket);
	debug_output("close socket = %d\n", nSocket);
}
bool CRecvMessThread::recvMsg(int nSocket)
{	
	int nRet;
	char szRecvBuf[256]={0};
	nRet = m_data->Recvn(nSocket, szRecvBuf, sizeof(PacketHeader));
	debug_output("recv msg data size = %d, socket = %d\n", nRet, nSocket);
	if(nRet <= 0)
	{
		debug_output("Error Recv msg failed ...... Socket = %d, OnlineClient = %d\n", nSocket, g_nCounter);
		return false;
	}
	else
	{
		PacketHeader* header = (PacketHeader*)szRecvBuf;
		debug_output(" Packet  Socket = %d, Size = %d, Type = %d\n", nSocket, header->nLength, header->nType);
		if(header->nType == HEARTBEAT_PING_PACKET)
		{
			PacketHeader heartbeat_res;
			heartbeat_res.nLength = 4;
			heartbeat_res.nFlags = 0;
			heartbeat_res.nType = HEARTBEAT_PONG_PACKET;

			if(sizeof(PacketHeader) != m_data->Sendn(nSocket, &heartbeat_res, sizeof(PacketHeader)))
			{
				return false;
			}
			modEpoll(nSocket, EPOLLIN);
		}
		else if(header->nType == ORDERINSERT_PACKET)
		{
			char content[sizeof(OrderInsertField) + 1] = {0};
			nRet = m_data->Recvn(nSocket, szRecvBuf, sizeof(OrderInsertField));

			RingBufferOrder* prb = RingBufferOrder::Instance();
			int64_t sequence = prb->ring_buffer.Next();
			OrderEvent* order = prb->ring_buffer.Get(sequence);
			order->SetValue(content);
			prb->ring_buffer.Publish(sequence);

			if(nRet != sizeof(OrderInsertField))
			{
				return false;
			}
			modEpoll(nSocket, EPOLLOUT);
		}
		else 
		{
			debug_output("Unrecognize Packet  Socket = %d, Type = %d\n", nSocket, header->nType);
			return false;
		}
	}
	return true;
}
/*struct timeval start, end;
int timeuse;
void time()
{
	gettimeofday(&end, NULL);
	timeuse = 1000000 * (end.tv_sec - start.tv_sec)+end.tv_usec - start.tv_usec;
	timeuse /= 1000;
	printf("time used: %d (ms)\n", timeuse);
}*/
void CRecvMessThread::Run()
{
//	pause();    /// 暂停线程
	int nfds , sock;
//	int nRet;
//	static bool bfirst = true;
	while(1)
	{
	//	printf("ewewewe\n");
		nfds = epoll_wait(epfd,events,5000, -1);
		for(int i=0; i<nfds; i++)
		{
			//debug_output("Epoll %d, i = %d, socket = %d\n", nfds, i, events[i].data.fd);
			if(events[i].events&EPOLLIN)
			{
				if((sock = events[i].data.fd) < 0)
					continue;
				if(g_nCounter > g_nMaxCounter)
				{
					modEpoll(sock, EPOLLIN);
					continue;
				}
				if(!recvMsg(sock))
				{
					delEpoll(sock);
					events[i].data.fd = -1;
					g_nCounter--;
				}
			}
			else if(events[i].events&EPOLLOUT)
			{
				if((sock = events[i].data.fd) < 0)
					continue;
			/*	if(bfirst)
				{
					gettimeofday(&start, NULL);
					bfirst = false;
				}*/
				//if(!m_threadPool->sendFileContent(sock))
				//{
				//	delEpoll(sock);
				//	events[i].data.fd = -1;
				//	g_nCounter--;	
				////	time();
				//	debug_output("Socket close ...... Socket = %d, OnlineCleint = %d\n", sock, g_nCounter);
				//}
				//else
				//{
				//	modEpoll(sock, EPOLLOUT);
				//	usleep(g_sleep);
				//}
			}
			else if(events[i].events&(EPOLLERR|EPOLLHUP))
			{
				if((sock = events[i].data.fd) < 0)
					continue;
				delEpoll(sock);
				events[i].data.fd = -1;
				g_nCounter--;
				debug_output("Error Client Accident Quit ...... Socket = %d, OnlineClient = %d\n", sock, g_nCounter);
			}
			else
			{
				sock = events[i].data.fd;
				delEpoll(sock);
				events[i].data.fd = -1;
				g_nCounter--;
				debug_output("Error Client Other Quit ...... Socket = %d, OnlineClient = %d\n", sock, g_nCounter);	
			}
		}
	}
}
